package tcpsubpub

import (
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/factory"
	"gitee.com/ymofen/gobase/gotask"
	"gitee.com/ymofen/gobase/gotask/chantask"
	"gitee.com/ymofen/gobase/subpub"
	"gitee.com/ymofen/kplbase/core/utils"
	"io"
	"sync/atomic"
)

type SubTcpServerSession struct {
	closedflag int32
	notify     subpub.ISubPub
	sessionid  string
	conf       gobase.StrMap

	pubque     string
	pubqueTask *chantask.ChanTask
	pubfun     func(topic string, args ...interface{}) int

	copybufflag int8

	operaRef *gobase.ObjectInnerRef
	opera    *TcpServerNode
}

func CreateSupTcpServerSession(args ...interface{}) (rval interface{}, err error) {
	plg := &SubTcpServerSession{
		sessionid: args[0].(string),
		conf:      args[1].(gobase.StrMap),
		notify:    args[2].(subpub.ISubPub),
	}
	return plg, nil
}

func (this *SubTcpServerSession) Open() error {
	if gobase.GoFunCatchException {
		defer gobase.DeferCatchPanic()
	}
	gotask.DefaultPatrolTask().CheckStart()
	return nil
}

func (this *SubTcpServerSession) PostData(args ...interface{}) error {
	return nil
}

func (this *SubTcpServerSession) innerPubQueue(topic string, args ...interface{}) int {
	action := args[0].(int)

	pubtask := this.pubqueTask
	if pubtask != nil {
		// action, sessionid, buf
		if action == subpub.ACTION_SERVER_CONN_RECV_BUF {
			if this.copybufflag == 1 {
				buf := args[2].([]byte)
				pubBuf := gobase.CloneBytes(buf, 0, 0)
				pubtask.PushTask(func(args ...interface{}) {
					if atomic.LoadInt32(&this.closedflag) == 1 {
						return
					}
					this.innerPub(topic, action, args[0], args[1])
				}, args[1], pubBuf)
			} else {
				pubtask.PushTask(func(args ...interface{}) {
					if atomic.LoadInt32(&this.closedflag) == 1 {
						return
					}
					this.innerPub(topic, args...)
				}, args...)
			}
		} else {
			pubtask.PushTask(func(args ...interface{}) {
				if atomic.LoadInt32(&this.closedflag) == 1 {
					return
				}
				this.innerPub(topic, args...)
			}, args...)
		}
		return 1
	} else {
		return 0
	}
}

func (this *SubTcpServerSession) innerPub(topic string, args ...interface{}) int {
	ex := this.notify
	if ex == nil {
		return 0
	}
	ex.Pub(this.sessionid, 0, args...)
	return 1
}

// 连接参数 connstr=&maxonline=0&network
//
//queflag:1, 排队发送
//copybuf:1, 发送时copy一份再发送,queflag:1时有效
func (this *SubTcpServerSession) Start() {
	//connstr := fmt.Sprintf("connstr=%s&user=%s&pass=%s&connrand=%s", this.conf.StringByName("connstr", ""),
	//	this.conf.StringByName("user", ""), this.conf.StringByName("pass", ""), this.conf.StringByName("connrand", ""))

	s0 := this.conf.StringByName("connstr", "")
	connf := utils.ParseConnString(s0)
	network := this.conf.StringByName("network", "")
	if len(network) == 0 {
		network = connf.Type
	}
	if len(network) == 0 {
		network = "tcp4"
	}

	connstr := fmt.Sprintf("network=%s&addr=%s",
		network, connf.Addr)

	this.operaRef = DefaultTcpServerGroup.CheckGetRef(connstr)
	this.opera = this.operaRef.GetObject().(*TcpServerNode)
	maxonline := this.conf.IntByName("maxonline", -1)
	if maxonline != -1 {
		this.opera.maxonline = maxonline
	}

	this.opera.ex.Sub(this.sessionid, "evt", func(id, topic string, args ...interface{}) (ok bool) {
		action := args[0].(int)

		if action == subpub.ACTION_SERVER_CONN_SEND_BUF {
			return false
		}

		this.pubfun(topic, args...)
		return true
	})
	//this.opera.ex.Sub(this.sessionid, "buf", func(id, topic string, args ...interface{}) (ok bool) {
	//	this.pubfun(pubTopic, args[0].([]byte))
	//	return true
	//})
	this.opera.checkStart()

	this.pubque = this.conf.StringByName("pubque", "")
	if len(this.pubque) > 0 {
		this.pubqueTask, _ = chantask.DefaultChanTaskHub.CheckGetChanTask(this.pubque, 1024, 0)
		this.pubfun = this.innerPubQueue
	} else {
		this.pubfun = this.innerPub
	}
}

func (this *SubTcpServerSession) Close() error {
	if !atomic.CompareAndSwapInt32(&this.closedflag, 0, 1) {
		return io.ErrClosedPipe
	}
	if this.operaRef != nil {
		this.opera.ex.Unsub(this.sessionid, "evt")
		this.opera = nil

		DefaultTcpServerGroup.ReleaseRef(this.operaRef)
		this.operaRef = nil
	}

	gotask.DefaultPatrolTask().DelTask(this.sessionid)

	this.notify.Pub(this.sessionid, 0, subpub.ACTION_CLOSE)
	this.notify = nil
	return nil
}

func init() {
	factory.RegisterCreateFunc("sub.tcpserver", CreateSupTcpServerSession)
}
